package com.atguigu.flink.state;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2023/3/1
 *
 *  TTL: time to live 存活时间。
 *          作用： 就是删除那些低频使用，过时的，僵尸状态，清理空间！
 *
 *      流式应用，需要 7 * 24 * 4 * 12
 *
 *
 *      累积水位
 */
public class Demo9_StateTTL
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        //状态TTL的配置对象
        StateTtlConfig ttlConfig = StateTtlConfig
            // 15秒是 处理时间的15秒，不支持事件时间。
            .newBuilder(Time.seconds(15))
            /*
                如何判断状态过期?
                    开启了ttl功能后，每一个状态，会额外添加一个 访问时间 的字段
                        s1:10 ,  2023-03-01 15:09:08

                    下次，这个状态被使用(读，写)的时候，使用当前时间和状态的 lastVistTime进行对比，如果差值超过 ttl，就会设置为过期，就会删除。
                    这个状态是一个僵尸状态，一直不被访问，此时在后台会有定期清理和检查的线程，帮你清理。
             */
            .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
            //状态过期了，不返回
            .neverReturnExpired()
            .build();

        env
            .socketTextStream("hadoop103", 8888)
            .map(new WaterSensorMapFunction())
            .keyBy(WaterSensor::getId)
            .process(new KeyedProcessFunction<String, WaterSensor, String>()
            {

                @Override
                public void open(Configuration parameters) throws Exception {
                    ValueStateDescriptor<Integer> descriptor = new ValueStateDescriptor<>("lastVc", Integer.class);
                    //开启ttl
                    descriptor.enableTimeToLive(ttlConfig);

                    sumVc = getRuntimeContext().getState(descriptor);
                }

                private ValueState<Integer> sumVc;
                @Override
                public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {

                    //取出 ValueState 存储的值
                    Integer sumVcNum = sumVc.value();

                   if (sumVcNum == null){
                       //当前是这个key的第一个WaterSensor
                       sumVc.update(value.getVc());

                   }else{

                       sumVc.update(sumVc.value() + value.getVc());

                   }
                   out.collect(ctx.getCurrentKey() + ":" +sumVc.value());

                }
            })
            .print();


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}
